package com.hwangrq.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;
import org.apache.dubbo.rpc.support.RpcUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author hwangrq
 */
@Activate(group = {CommonConstants.CONSUMER})
public class TpMonitorFilter implements Filter {

    /** methodName,[currentTime, costTime] */
    private static Map<String, ConcurrentHashMap<Long,Long>> methodCostMap = new ConcurrentHashMap<>();

    private static ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor();

    static {
        // 5s 统计最近一分钟tp90，tp99，打印出结果
        timer.scheduleWithFixedDelay(() -> {
            //遍历 methodCostMap
            for (Map.Entry<String, ConcurrentHashMap<Long, Long>> entry : methodCostMap.entrySet()) {
                //取出请求时间
                ConcurrentHashMap<Long, Long> costTimeMap = entry.getValue();

                Iterator<Map.Entry<Long, Long>> iterator = costTimeMap.entrySet().iterator();

                //把请求时间存储list中
                List<Long> costList = new ArrayList<>();

                while (iterator.hasNext()) {
                    Map.Entry<Long, Long> costs = iterator.next();
                    // 如果超过1分钟则删除记录
                    if (System.currentTimeMillis() - costs.getKey() > 60000) {
                        iterator.remove();
                    }else{
                        costList.add(costs.getValue());
                    }
                }

                Collections.sort(costList);
                int requestCount = costList.size();
                long tp90 = costList.get((int)(requestCount * 0.9));
                long tp99 = costList.get((int)(requestCount * 0.99));
                System.out.printf("方法【%s】，请求总数：%s，tp90: %s, tp99: %s %n", entry.getKey(), requestCount, tp90, tp99);
            }
        }, 0, 5, TimeUnit.SECONDS);
    }


    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String methodName = invocation.getMethodName();
        final boolean isAsync = RpcUtils.isAsync(invoker.getUrl(), invocation);
        //异步则不处理
        if (!isAsync) {
            long startTime = System.currentTimeMillis();
            try {
                // 执行方法
                return invoker.invoke(invocation);
            } finally {
                long costTime = System.currentTimeMillis() - startTime;
                //判断方法是否存在于map中
                if(!methodCostMap.containsKey(methodName)){
                    methodCostMap.put(methodName, new ConcurrentHashMap<>());
                }else{
                    methodCostMap.get(methodName).put(System.currentTimeMillis(), costTime);
                }
            }
        }
        return invoker.invoke(invocation);
    }
}
